package com.hery.utils;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

@Slf4j
public class MysqlCDCDeserializationSchema implements DebeziumDeserializationSchema<String> {

    public enum Status {

        AFTER("after"),
        BEFORE("before");

        private final String code;

        Status(String code) {
            this.code = code;
        }

        public String code() {
            return this.code;
        }
    }


    /**
     * create: 新增记录，只有 after
     * delete：只有 before， 没有 after
     * update: after + before
     * select：
     */
    public void deserialize(SourceRecord record, Collector<String> out) {

        Envelope.Operation op = Envelope.operationFor(record);
        Struct value = (Struct)record.value();

        if (op != Envelope.Operation.CREATE && op != Envelope.Operation.READ) {
            if (op == Envelope.Operation.DELETE) {
                log.debug("delete:{}", value);
                String before = this.extractBeforeRow(value);
                out.collect(before);
            } else {
                log.debug("update:{}", value);
                String beforeAndAfter = this.extractBeforeAndAfterRow(value);
                out.collect(beforeAndAfter);
            }
        } else {
            log.debug("create:{}", value);
            String after = this.extractAfterRow(value);
            out.collect(after);
        }

    }

    private String extractAfterRow(Struct value) {
        Struct after = value.getStruct(Status.AFTER.code());
        JSONObject jsonObject = new JSONObject();
        jsonObject.put(Status.AFTER.code(), parseRecord(after));
        return jsonObject.toJSONString();
    }

    private String extractBeforeRow(Struct value) {
        Struct before = value.getStruct(Status.BEFORE.code());
        JSONObject jsonObject = new JSONObject();
        jsonObject.put(Status.BEFORE.code(), parseRecord(before));
        return jsonObject.toJSONString();
    }

    private String extractBeforeAndAfterRow(Struct value) {

        Struct before = value.getStruct(Status.BEFORE.code());
        Struct after = value.getStruct(Status.AFTER.code());

        JSONObject jsonObject = new JSONObject();
        jsonObject.put(Status.BEFORE.code(), parseRecord(before));
        jsonObject.put(Status.AFTER.code(), parseRecord(after));
        return jsonObject.toJSONString();
    }

    private JSONObject parseRecord(Struct after) {
        JSONObject jo = new JSONObject();
        for (Field field : after.schema().fields()) {
            switch ((field.schema()).type()) {
                case INT8:
                    int int8 = after.getInt8(field.name());
                    jo.put(field.name(), int8);
                    break;
                case INT16:
                    int int16 = after.getInt16(field.name());
                    jo.put(field.name(), int16);
                    break;
                case INT32:
                    int int32 = after.getInt32(field.name());
                    jo.put(field.name(), int32);
                    break;
                case INT64:
                    Long resultInt = after.getInt64(field.name());
                    jo.put(field.name(), resultInt);
                    break;
                case FLOAT32:
                    Float resultFloat32 = after.getFloat32(field.name());
                    jo.put(field.name(), resultFloat32);
                    break;
                case FLOAT64:
                    Double resultFloat64 = after.getFloat64(field.name());
                    jo.put(field.name(), resultFloat64);
                    break;
                case BYTES:
                     byte[] resultByte = after.getBytes(field.name());
                     jo.put(field.name(), new String(resultByte));
                    break;
                case STRING:
                    String resultStr = after.getString(field.name());
                    jo.put(field.name(), resultStr);
                    break;
                default:
                    log.warn("未知的数据类型：{}", (field.schema()).type());
                    jo.put(field.name(), after);
            }
        }

        return jo;
    }

    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}
